package com.smsc.headend.task.engine.service.collection.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.lang.Tuple;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.base.Preconditions;
import com.smsc.headend.common.utils.CollectionIntervalsUtils;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.scheme.entity.ConfigAutoRecollectTask;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.proto.TaskProto;
import com.smsc.headend.task.engine.exception.UDISTaskExecutorException;
import com.smsc.headend.task.engine.mapper.ConfigAutoRecollectTaskMapper;
import com.smsc.headend.task.engine.po.CollectDataDetailBo;
import com.smsc.headend.task.engine.po.RecollectInfo;
import com.smsc.headend.task.engine.service.collection.RecollectTaskService;
import com.smsc.headend.task.engine.service.task.TaskService;
import com.smsc.headend.task.engine.utils.TaskParamUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang.time.DateUtils;
import org.assertj.core.util.Lists;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.annotation.CacheConfig;
import org.springframework.cache.annotation.CachePut;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
@CacheConfig(cacheNames = {"ute.recollect"})
public class RecollectTaskServiceImpl extends ServiceImpl<ConfigAutoRecollectTaskMapper, ConfigAutoRecollectTask> implements RecollectTaskService {
    public static final String RECOLLECT_TASK_SEQ = "recollectTaskSeq";
    @Autowired
    RedisUtils redisUtils;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    @Lazy
    RecollectTaskService self;
    @Autowired
    TaskService taskService;
    @Autowired
    ConfigAutoRecollectTaskMapper configAutoRecollectTaskMapper;

    @Override
    public boolean changeDetailIfRequired(Task task, AtomicTask atomicTask) throws UDISTaskExecutorException {
        Map<String, Object> parameterMap = TaskParamUtils.getParameterMap(task, atomicTask);
        Long recollectSeq = Convert.toLong(parameterMap.get(RECOLLECT_TASK_SEQ));
        String targetPoint = Convert.toStr(parameterMap.get("targetPoint"));
        if (StringUtils.isEmpty(targetPoint)) {
            throw new UDISTaskExecutorException(UDISTaskExecutorException.TASK_EXPIRED);
        }
        String changedRecollectDetail = self.getChangedRecollectDetail(recollectSeq);
        if (StringUtils.isEmpty(changedRecollectDetail)) {
            return false;
        }

        byte[] changed = ByteUtils.hexStringToByte(changedRecollectDetail);
        byte[] target = ByteUtils.hexStringToByte(targetPoint);
        Preconditions.checkArgument(changed.length == target.length);
        for (int i = 0; i < changed.length; i++) {
            byte temp = (byte) ((~changed[i]));
            byte t2 = (byte) (target[i] & temp);
            target[i] &= ~t2;
        }
        String targetPointAfter = ByteUtils.byteToHexString(target).replaceAll(" ", "");
        if (!targetPointAfter.equalsIgnoreCase(targetPoint)) {
            log.info("recollect: point change,seq={},meterId:{},before:{},after:{}", recollectSeq, task.getMeterId(), targetPoint, targetPointAfter);
            Map<String, Object> map = JSONUtil.toBean(task.getParamJson(), HashMap.class);
            map.put("targetPoint", targetPointAfter);
            map.put("executeTime", "1");
            task.setParamJson(JSONUtil.toJsonStr(map));
            return taskService.updateTask(task);
        }
        return false;
    }

    @Override
    public void rescheduleRecollectTask(Task task, CollectDataDetailBo collectDataDetailBo, Integer executedTimes) {
        Task duplicate = new Task();
        BeanUtils.copyProperties(task, duplicate);

        Map<String, String> paramMap = JSONUtil.toBean(duplicate.getParamJson(), Map.class);
        paramMap.put("executeTime", Convert.toStr(executedTimes + 1));
        paramMap.put("targetPoint", collectDataDetailBo.getUpdatedDataDetail());
        String topic = decideTopicByTimes(executedTimes);
        if (StringUtils.isEmpty(topic)) {
            return;
        }
        duplicate.setTaskNo(UUID.randomUUID().toString());
        duplicate.setParamJson(JSONUtil.toJsonStr(paramMap));
        Integer partitionCount = 3;
        Long partition = (duplicate.getMeterId() == null ? duplicate.getComId() : duplicate.getMeterId()) % partitionCount;
        Preconditions.checkNotNull(partition);
        Preconditions.checkNotNull(duplicate);
        log.info("send recollect task: topic= {}, {}", topic, JSONUtil.toJsonStr(paramMap));
        kafkaTemplate.send(topic, Convert.toInt(partition), SystemClock.now(), UUID.randomUUID().toString(), ProtoBufferUtil.toProtoBuffer(duplicate, TaskProto.Task.class).toByteArray());
    }

    private String decideTopicByTimes(Integer executedTimes) {
        switch (executedTimes) {
            case 1:
            case 2:
                return TaskKafkaTopic.TASK_DELAY60_TO_UTE;
            case 3:
            case 4:
                return TaskKafkaTopic.TASK_DELAY240_TO_UTE;
            default:
                return null;
        }
    }

    @Override
    public int checkRecollectTask(Long meterId, Long interval, Set<Long> absentPointSet, Long lpDataItemId) {
        Map<Tuple, Set<Long>> periodTimeSetMap = getRecollectMap(absentPointSet, interval);
        periodTimeSetMap.entrySet().stream().forEach(tupleSetEntry -> {
            RecollectInfo recollectInfo = new RecollectInfo(meterId, tupleSetEntry.getKey().get(0), tupleSetEntry.getKey().get(1), tupleSetEntry.getValue(), lpDataItemId, interval);
            String rcltJson = JSONUtil.toJsonStr(recollectInfo);
            log.debug("recollect: send recollect info, {}", rcltJson);
            kafkaTemplate.send(TaskKafkaTopic.RECOLLECT_TASK_GEN, rcltJson.getBytes(StandardCharsets.UTF_8));
        });
        return periodTimeSetMap.keySet().size();
    }

    @Override
    public List<ConfigAutoRecollectTask> getRecollectTask(long lpDataItemId, long startTv, long endTv, long meterId) {
        log.info("recollect: query recollect task,meterId:{}, lpDataItemId:{},tvStart:{}", meterId, lpDataItemId, endTv);
        LambdaQueryWrapper<ConfigAutoRecollectTask> queryWrapper = Wrappers.<ConfigAutoRecollectTask>lambdaQuery().eq(ConfigAutoRecollectTask::getLpDataItemId, lpDataItemId).eq(ConfigAutoRecollectTask::getTvStart, startTv)
                .eq(ConfigAutoRecollectTask::getTvEnd, endTv).eq(ConfigAutoRecollectTask::getMeterId, meterId);
        List<ConfigAutoRecollectTask> configAutoRecollectTasks = this.configAutoRecollectTaskMapper.selectList(queryWrapper);
        return CollectionUtils.isEmpty(configAutoRecollectTasks) ? null : configAutoRecollectTasks;
    }

    @Override
    @CachePut(key = "'rclt:detail.' + #configAutoRecollectTask.seq", unless = "#result == null")
    public String saveChangedRecollectDetail(ConfigAutoRecollectTask configAutoRecollectTask, Set<Long> absentPoints) {
        log.info("recollect: changeDetail,seq:{}, absentPoints:{}", configAutoRecollectTask.getSeq(), JSONUtil.toJsonStr(absentPoints));
        Long tvStart = configAutoRecollectTask.getTvStart();
        Long tvEnd = configAutoRecollectTask.getTvEnd();
        int interval = configAutoRecollectTask.getCaptureInterval().intValue();
        String dayDetail = CollectionIntervalsUtils.pointDataTransferToHex(tvStart, tvEnd, interval, Lists.newArrayList(absentPoints), Boolean.TRUE);
        String calculatedDetail;
        try {
            calculatedDetail = CollectionIntervalsUtils.calAutoRecollectDetailFromDayIntegrity(dayDetail, interval, tvStart, tvEnd);
        } catch (IllegalArgumentException e) {
            log.error("RecollectTaskIntervalNotEvenToCurrentInterval:meterId:{},seq:{},detail:{},currentInterval:{}", configAutoRecollectTask.getMeterId(), configAutoRecollectTask.getSeq(), dayDetail, interval);
            calculatedDetail = ByteUtils.byteToHexString(new byte[(int) ((tvEnd - tvStart) / interval)]);
        }
        return calculatedDetail.equalsIgnoreCase(configAutoRecollectTask.getDataDetail()) ? null : calculatedDetail;
    }

    @Override
    @Cacheable(key = "'rclt:detail.' + #seq", unless = "#result == null")
    public String getChangedRecollectDetail(Long seq) {
        return null;
    }

    @Override
    public int generateRecollectTask(Long meterId, Long startTime, long endTime, Long lpDataItemId, Integer intervalTime) {
        List<Tuple> startEndTuples = getStartEnd(startTime, endTime, intervalTime);
        Preconditions.checkArgument(!org.springframework.util.CollectionUtils.isEmpty(startEndTuples));
        startEndTuples.stream().forEach(t -> this.generateRecollectTaskByStartEnd(t, meterId, lpDataItemId, intervalTime));
        return 1;
    }


    private List<Tuple> getStartEnd(Long startTime, Long endTime, Integer intervalTime) {
        List<Tuple> startEndTuples = new ArrayList<>();
        if (intervalTime < 86400) {
            Date startDate = CollectionIntervalsUtils.getRecollectTaskStartDate(com.google.common.collect.Lists.newArrayList(startTime, endTime));
            Long tempTime = startDate.getTime();

            while (tempTime < endTime * DateUtils.MILLIS_PER_SECOND) {
                startEndTuples.add(new Tuple(tempTime, tempTime + 8 * DateUtils.MILLIS_PER_HOUR));
                tempTime += 8 * DateUtils.MILLIS_PER_HOUR;
            }
        }

        if (intervalTime >= 86400) {
            Set<Long> absentPointList = getAbsentDayOrMonthPoint(startTime, endTime, intervalTime);
            Optional.ofNullable(absentPointList).orElse(Collections.emptySet()).stream()
                    .filter(tv -> !tv.equals(startTime)) //过滤掉起点，因为已经采集到或者在之前的补抄任务中
                    .forEach(tv -> {
                        Long startTv = tv - 12 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;
                        Long endTv = tv + 12 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;
                        startEndTuples.add(new Tuple(startTv * DateUtils.MILLIS_PER_SECOND, endTv * DateUtils.MILLIS_PER_SECOND));
                    });
            //当最后一个点在零点整点时， CollectionIntervalsUtils.getAbsentPointList方法 计算缺点会去除最后一个点。 这里主动添加到list中
            String endTimeFormat = DateFormatUtils.format(new Date(endTime * DateUtils.MILLIS_PER_SECOND), DateFormatUtils.ISO_DATETIME_FORMAT.getPattern());
            if (endTimeFormat.endsWith("00:00:00")) {
                Long startTv = endTime - 12 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;
                Long endTv = endTime + 12 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;
                startEndTuples.add(new Tuple(startTv * DateUtils.MILLIS_PER_SECOND, endTv * DateUtils.MILLIS_PER_SECOND));
            }
        }
        return startEndTuples;
    }

    private Set<Long> getAbsentDayOrMonthPoint(Long startTime, Long endTime, Integer intervalTime) {
        Date startDate = DateUtils.truncate(new Date(startTime * 1000), Calendar.DAY_OF_MONTH);
        Date endDate = DateUtils.truncate(new Date(endTime * 1000), Calendar.DAY_OF_MONTH);
        Set<Long> hashSet = new HashSet<>();
        Date tmpDate = startDate;
        do {
            hashSet.add(intervalTime == 86400L ? tmpDate.getTime() / 1000
                    : DateUtils.truncate(tmpDate, Calendar.MONTH).getTime() / 1000);
            tmpDate = intervalTime == 86400L ? DateUtils.addDays(tmpDate, 1)
                    : DateUtils.addMonths(tmpDate, 1);
        } while (tmpDate.before(endDate));
        return hashSet;
    }

    /**
     * 强制生成补抄记录
     *
     * @param startEndTime {@link Tuple}
     *                     Minute Profile:
     *                     startEndTime.get(0) {startTimestamp -- milliseconds of recollect task start time;  00:00 08:00 16:00}
     *                     startEndTime.get(1) {endTimestamp -- milliseconds of recollect task end time;   08:00 16:00 24:00}
     *                     Day Profile
     *                     startEndTime.get(0) {startTimestamp -- milliseconds of recollect task start time;  e.q: 2021-11-01 12:00:00}
     *                     startEndTime.get(1) {endTimestamp -- milliseconds of recollect task end time;   2021-11-02 12:00:00}
     *                     Month Profile
     *                     startEndTime.get(0) {startTimestamp -- milliseconds of recollect task start time;  e.q: 2021-10-31 12:00:00}
     *                     startEndTime.get(1) {endTimestamp -- milliseconds of recollect task end time;   2021-11-01 12:00:00}
     * @param meterId
     * @param lpDataItemId
     * @param intervalTime
     */
    private void generateRecollectTaskByStartEnd(Tuple startEndTime, Long meterId, Long lpDataItemId, Integer intervalTime) {
        Preconditions.checkNotNull(startEndTime.get(0));
        Preconditions.checkNotNull(startEndTime.get(1));
        Long startTime = Convert.toLong(startEndTime.get(0)) / 1000;
        Long endTime = Convert.toLong(startEndTime.get(1)) / 1000;

        Integer dataPoints = intervalTime >= 86400 ? 1 : Convert.toInt(8 * DateUtils.MILLIS_PER_HOUR / (DateUtils.MILLIS_PER_SECOND * intervalTime));
        Integer byteLen = Convert.toInt((endTime - startTime) / (intervalTime.longValue() * 8));
        String detail = intervalTime >= 86400 ? "00" : ByteUtils.byteToHexString(new byte[byteLen]).replaceAll(" ", "");
        ConfigAutoRecollectTask autoRecollectTask = new ConfigAutoRecollectTask();
        autoRecollectTask.setMeterId(meterId);
        autoRecollectTask.setLpDataItemId(lpDataItemId);
        autoRecollectTask.setTvStart(startTime);
        autoRecollectTask.setTvEnd(endTime);
        autoRecollectTask.setDataPoints(dataPoints);
        autoRecollectTask.setCaptureInterval(intervalTime.longValue());
        autoRecollectTask.setTvDataHigh(endTime);
        autoRecollectTask.setTvDataLow(startTime);
        autoRecollectTask.setDataDetail(detail);
        autoRecollectTask.setExecutedTimes(0);
        autoRecollectTask.setStatus(ConfigAutoRecollectTask.Status.Waiting.getCode());
        log.info("recollect save: start:{}, end:{}, lp:{}, interval:{}, detail:{}", DateFormatUtils.format(new Date(autoRecollectTask.getTvStart() * 1000), DateFormatUtils.ISO_DATETIME_FORMAT.getPattern())
                , DateFormatUtils.format(new Date(autoRecollectTask.getTvEnd() * 1000), DateFormatUtils.ISO_DATETIME_FORMAT.getPattern()),
                autoRecollectTask.getLpDataItemId(), autoRecollectTask.getCaptureInterval(), autoRecollectTask.getDataDetail());
        kafkaTemplate.send(TaskKafkaTopic.RECOLLECT_TASK_DATA_SAVE, Convert.toStr(meterId), JSONUtil.toJsonStr(autoRecollectTask).getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public void createRecollectTask(RecollectInfo recollectInfo) {
        List<ConfigAutoRecollectTask> configAutoRecollectTask = this.getRecollectTask(recollectInfo.getLpDataItemId(), recollectInfo.getStartTv(), recollectInfo.getEndTv(), recollectInfo.getMeterId());
        if (noRecollectTaskExits(configAutoRecollectTask)) {
            this.addAutoRecollectPlan(recollectInfo.getMeterId(), recollectInfo.getLpDataItemId(), recollectInfo.getStartTv()
                    , recollectInfo.getEndTv(), Lists.newArrayList(recollectInfo.getAbsentPoints()), Convert.toInt(recollectInfo.getInterval()), recollectInfo.getAbsentPoints().size(), null);
        } else {
            List<ConfigAutoRecollectTask> collectingTask = configAutoRecollectTask.stream().filter(t -> ConfigAutoRecollectTask.Status.Waiting.getCode().equals(t.getStatus()) ||
                    ConfigAutoRecollectTask.Status.WaitingForReCollect.getCode().equals(t.getStatus()) ||
                    ConfigAutoRecollectTask.Status.ReCollecting.getCode().equals(t.getStatus())).collect(Collectors.toList());
            collectingTask.stream().forEach(o -> this.saveChangedRecollectDetail(o, recollectInfo.getAbsentPoints()));
        }
    }

    private boolean noRecollectTaskExits(List<ConfigAutoRecollectTask> configAutoRecollectTask) {
        if (CollectionUtils.isEmpty(configAutoRecollectTask)) {
            return true;
        }

        return !configAutoRecollectTask.stream().filter(t -> ConfigAutoRecollectTask.Status.Waiting.getCode().equals(t.getStatus()) ||
                ConfigAutoRecollectTask.Status.WaitingForReCollect.getCode().equals(t.getStatus()) ||
                ConfigAutoRecollectTask.Status.ReCollecting.getCode().equals(t.getStatus())).findAny().isPresent();
    }


    private Map<Tuple, Set<Long>> getRecollectMap(Set<Long> absentPointSet, Long interval) {
        Preconditions.checkArgument(CollectionUtils.isNotEmpty(absentPointSet));
        if (interval <= 86400) {
            return getMinuteRecollectMap(absentPointSet);
        } else {
            return getDayOrMonthRecollectMap(absentPointSet);
        }
    }

    private Map<Tuple, Set<Long>> getDayOrMonthRecollectMap(Set<Long> absentPointSet) {
        Map<Tuple, Set<Long>> resultMap = new HashMap<>();
        for (Long tv : absentPointSet) {
            Long startTv = tv - 12 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;
            Long endTv = tv + 12 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;
            Tuple cursor = new Tuple(startTv, endTv);
            if (!resultMap.containsKey(cursor)) {
                resultMap.put(cursor, new HashSet<>());
            }
            resultMap.get(cursor).add(tv);
        }
        return resultMap;
    }

    private Map<Tuple, Set<Long>> getMinuteRecollectMap(Set<Long> absentPointSet) {
        ArrayList<Long> tvs = Lists.newArrayList(absentPointSet);
        tvs.sort(Long::compareTo);
        Date startDate = CollectionIntervalsUtils.getRecollectTaskStartDate(tvs);

        long startTv = startDate.getTime() / DateUtils.MILLIS_PER_SECOND;
        long endTv = startTv + 8 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;

        Map<Tuple, Set<Long>> resultMap = new HashMap<>();
        for (Long tv : tvs) {
            if (tv >= endTv) {
                startTv = endTv;
                endTv = startTv + 8 * DateUtils.MILLIS_PER_HOUR / DateUtils.MILLIS_PER_SECOND;
            }
            Tuple cursor = new Tuple(startTv, endTv);
            if (!resultMap.containsKey(cursor)) {
                resultMap.put(cursor, new HashSet<>());
            }
            resultMap.get(cursor).add(tv);
        }
        return resultMap;
    }


    public Integer addAutoRecollectPlan(Long meterId, Long lpDataItemId, Long startTime, Long endTime, List<Long> absentPointTvList, Integer intervalTime, Integer dataPoints, String absentPointsData) {
        if (startTime > SystemClock.now() / 1000) {
            log.info("recollect: add plan tv too recent. meterId={},lp:{},tv:{} -> {}", meterId, lpDataItemId, startTime, endTime);
            return 0;
        }
        ConfigAutoRecollectTask autoRecollectTask = new ConfigAutoRecollectTask();
        autoRecollectTask.setMeterId(meterId);
        autoRecollectTask.setLpDataItemId(lpDataItemId);
        autoRecollectTask.setTvStart(startTime);
        autoRecollectTask.setTvEnd(endTime);
        autoRecollectTask.setDataPoints(dataPoints);
        autoRecollectTask.setCaptureInterval(intervalTime.longValue());
        autoRecollectTask.setTvDataHigh(absentPointTvList == null ? endTime : absentPointTvList.stream().max(Long::compareTo).get());
        autoRecollectTask.setTvDataLow(absentPointTvList == null ? startTime : absentPointTvList.stream().min(Long::compareTo).get());
        String detail = absentPointsData == null ? CollectionIntervalsUtils.pointDataTransferToHex(startTime, endTime, intervalTime, absentPointTvList, Boolean.TRUE) : absentPointsData;
        detail = CollectionIntervalsUtils.calAutoRecollectDetailFromDayIntegrity(detail, intervalTime, startTime, endTime);
        autoRecollectTask.setDataDetail(detail);
        autoRecollectTask.setExecutedTimes(0);
        autoRecollectTask.setStatus(ConfigAutoRecollectTask.Status.Waiting.getCode());
        kafkaTemplate.send(TaskKafkaTopic.RECOLLECT_TASK_DATA_SAVE, JSONUtil.toJsonStr(autoRecollectTask).getBytes(StandardCharsets.UTF_8));
        log.info("recollect: created, meterId={}, seq={}, detail={}", meterId, autoRecollectTask.getSeq(), detail);
        return 1;
    }

    @Override
    @Transactional
    public CollectDataDetailBo handleRecollect(List<RdMeterReadsLp> readsLpList, String recollectTaskSeq, String targetPoint, Integer executeTime, String interval) {
        Set<Long> collectedPoints = new HashSet<>();
        if (!CollectionUtils.isEmpty(readsLpList)) {
            for (RdMeterReadsLp rdMeterReadsLp : readsLpList) {
                collectedPoints.add(rdMeterReadsLp.getTv());
            }
        }
        CollectDataDetailBo collectDataDetailBo = new CollectDataDetailBo();

        Long itvNumber = Convert.toLong(interval);
        String updatedDetail = targetPoint;
        Integer leftPointNumber = 0;
        if (!CollectionUtils.isEmpty(readsLpList)) {
            if (itvNumber >= DateUtils.MILLIS_PER_DAY / 1000) {
                updatedDetail = collectedPoints.isEmpty() ? updatedDetail : "01";
                leftPointNumber = collectedPoints.isEmpty() ? 1 : 0;
                collectDataDetailBo.setHitPointTv(collectedPoints);
            } else {
                byte[] combinedResult = CollectionIntervalsUtils.calculateMinuteCollectDetail(targetPoint, collectedPoints, itvNumber);
                byte[] targetBytes = ByteUtils.hexStringToByte(targetPoint);
                Set<Long> getHitTvPoints = CollectionIntervalsUtils.calculateRecollectHitTimestamp(combinedResult, targetBytes, collectedPoints, interval);
                updatedDetail = ByteUtils.byteToHexString(combinedResult).replaceAll(" ", "");
                leftPointNumber = CollectionIntervalsUtils.getZeroPointCount(updatedDetail);
                collectDataDetailBo.setHitPointTv(getHitTvPoints);
            }
        } else {
            leftPointNumber = null;
            updatedDetail = null;
            collectDataDetailBo.setHitPointTv(null);
        }

        ConfigAutoRecollectTask updater = new ConfigAutoRecollectTask();
        updater.setSeq(Long.valueOf(recollectTaskSeq));
        updater.setDataPoints(leftPointNumber);
        updater.setDataDetail(updatedDetail);
        updater.setTvLastColl(SystemClock.now() / 1000);
        updater.setStatus(leftPointNumber != null && leftPointNumber == 0 ? ConfigAutoRecollectTask.Status.Complete.getCode() :
                executeTime >= 4 ? ConfigAutoRecollectTask.Status.AutoRecollectFail.getCode() : ConfigAutoRecollectTask.Status.WaitingForReCollect.getCode());
        log.debug("recollect: logSeq={}, leftNumber={}, collectedPoints={}, updated={}", recollectTaskSeq, leftPointNumber, collectedPoints.size());
        collectDataDetailBo.setLeftPointCount(leftPointNumber);
        collectDataDetailBo.setUpdatedDataDetail(updatedDetail);
        kafkaTemplate.send(TaskKafkaTopic.RECOLLECT_TASK_UPDATE, JSONUtil.toJsonStr(updater).getBytes(StandardCharsets.UTF_8));
        return collectDataDetailBo;
    }


    @Override
    public void persistRecollectTask(List<ConfigAutoRecollectTask> dtoList) {
        long startTime = SystemClock.now();
        int sizeInit = dtoList.size();
        Map<Long, List<ConfigAutoRecollectTask>> collect = dtoList.stream().filter(c -> c.getSeq() != null).collect(Collectors.groupingBy(d -> d.getSeq()));
        List<ConfigAutoRecollectTask> distinctRecollectTask = collect.entrySet().stream().map(entry -> entry.getValue().stream().reduce((max, c) -> max.getTvLastColl() < c.getTvLastColl() ? max : c).get()).collect(Collectors.toList());
        int filteredSize = distinctRecollectTask.size();
        this.executeBatch(distinctRecollectTask, (sqlSession, recollectTask) -> {
            sqlSession.update("com.smsc.headend.task.engine.mapper.ConfigAutoRecollectTaskMapper.updateTask", recollectTask);
        });
        log.info("saveSize:{},filteredSize:{},cost:{}ms", sizeInit, filteredSize, SystemClock.now() - startTime);
    }

}
